/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.servicemix.audit.jcr;

import static org.apache.servicemix.audit.jcr.SlingUtils.getNodeCorrectName;

import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.GregorianCalendar;
import java.util.Locale;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.jbi.messaging.MessageExchange;
import javax.jbi.messaging.MessagingException;
import javax.jbi.messaging.NormalizedMessage;
import javax.jcr.ItemExistsException;
import javax.jcr.Node;
import javax.jcr.PathNotFoundException;
import javax.jcr.RepositoryException;
import javax.jcr.Session;
import javax.jcr.ValueFormatException;
import javax.jcr.lock.LockException;
import javax.jcr.nodetype.ConstraintViolationException;
import javax.jcr.version.VersionException;
import javax.xml.transform.TransformerException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.jackrabbit.value.DateValue;
import org.apache.servicemix.jbi.jaxp.SourceTransformer;
import org.apache.servicemix.jbi.util.MessageUtil;

/**
 * 
 * Class for processing message exchange based on correlation id
 * 
 * @author vkrejcirik
 * 
 */
public class SlingJcrAuditorStrategy implements JcrAuditorStrategy {

    private static final SourceTransformer TRANSFORMER = new SourceTransformer();
    private static final Log LOG = LogFactory.getLog(SlingJcrAuditorStrategy.class);

    // let's time slice our message exchange archive on an hourly basis
    private static final DateFormat FORMAT = new SimpleDateFormat("yyyyMMddhha", Locale.ENGLISH);

    // init the repository?
    private AtomicBoolean init = new AtomicBoolean(false);

    public void processExchange(MessageExchange messageExchange, Session session) throws ItemExistsException, PathNotFoundException,
            VersionException, ConstraintViolationException, LockException, RepositoryException, MessagingException, TransformerException {

        //only once, when init is false -> set to true
        if (init.compareAndSet(false, true)) {
            init(session);
        }

        String endpointName = "NO_ENDPOINT_NAME";
        String serviceName = "NO_SERVICE_NAME";
        
        
        if (messageExchange.getEndpoint() != null && messageExchange.getEndpoint().getEndpointName() != null)
            endpointName  = messageExchange.getEndpoint().getEndpointName();
        
        if (messageExchange.getService() != null && messageExchange.getService().getLocalPart() != null)
            serviceName = messageExchange.getService().getLocalPart();
         
        //create unique endpoints name
        String uniqueEndpointName = endpointName + ":" + serviceName;
        
        //node for new endpoint
        Node newNode = null;
        
        // get endpoint node for this exchange
        if (endpointName != null && serviceName != null) {
            
            Node endpointNode = getEndpointNode(session, uniqueEndpointName);

            // unique endpoint name
            endpointNode.setProperty("endpoint", (String) uniqueEndpointName);
            endpointNode.setProperty("endpointId", SlingUtils.getNodeCorrectName((String) uniqueEndpointName));
            
            // get correct name of node
            String exId = SlingUtils.getNodeCorrectName((String) messageExchange.getExchangeId());

            // add this exchange to the endpoint node, if not exists yet
            try {
                endpointNode.getNode(exId);
            } catch (PathNotFoundException e) {
                newNode = endpointNode.addNode(exId);

                newNode.setProperty("created", new DateValue(new GregorianCalendar()));
                newNode.setProperty("exchangeId", messageExchange.getExchangeId());
            }
        }

        // construct exchange node
        Node node = getNodeForExchange(messageExchange, session);

        String path = AuditContstants.MESSAGE_FLOWS_RESOURCE_TYPE;
        
        path += "/";
        path += node.getParent().getParent().getName();
        path += "/";
        path += node.getParent().getName();
        path += "/";
        path += node.getName();
        
        if (newNode != null) {
            newNode.setProperty("pathToResource", path);
        }   
            
        node.setProperty("pathToResource", path);
        
        if (messageExchange.getStatus() != null)
            node.setProperty("exchangeStatus", messageExchange.getStatus().toString());
        
        // icon for status
        if (messageExchange.getStatus().toString().equals("Done"))
            node.setProperty("statusIcon", AuditContstants.STATUS_DONE_ICON);
        else
            node.setProperty("statusIcon", AuditContstants.STATUS_ERROR_ICON);
        
        if (messageExchange.getPattern() != null)
            node.setProperty("pattern", messageExchange.getPattern().toString());

        //icon for MEP
        if (messageExchange.getPattern().toString().endsWith("in-out"))
            node.setProperty("mepIcon", AuditContstants.MEP_INOUT_ICON);
        else
            node.setProperty("mepIcon",AuditContstants.MEP_IN_ICON);
        

        node.setProperty("endpointId", SlingUtils.getNodeCorrectName(uniqueEndpointName));
        
        if (messageExchange.getEndpoint() != null) {
            node.setProperty("endpoint", uniqueEndpointName);
        }

        if (messageExchange.getService() != null) {
            node.setProperty("service", messageExchange.getService().toString());
        }

        for (Object key : messageExchange.getPropertyNames()) {
            String name = (String) key;
            node.setProperty(getNodeCorrectName(name), messageExchange.getProperty(name).toString());
        }

        addNormalizedMessages(node, messageExchange);
        node.setProperty("updated", new DateValue(new GregorianCalendar()));
    }

    private void init(Session session) throws ValueFormatException, VersionException, LockException, ConstraintViolationException,
            RepositoryException {
        LOG.info("Initializing the Sling repository");
        Node content = createOrGet(session.getRootNode(), "content", "");
        Node servicemix = createOrGet(content, "servicemix", "servicemix/dashboard");
        createOrGet(servicemix, "config", "servicemix/config");
        session.save();
    }

    /**
     * 
     * Get node for given Exchange
     * 
     * @param exchange
     * @param session
     * @return Node
     * @throws ItemExistsException
     * @throws PathNotFoundException
     * @throws VersionException
     * @throws ConstraintViolationException
     * @throws LockException
     * @throws RepositoryException
     */
    private Node getNodeForExchange(MessageExchange exchange, Session session) throws ItemExistsException, PathNotFoundException,
            VersionException, ConstraintViolationException, LockException, RepositoryException {

        // exchange id
        String id = exchange.getExchangeId();

        // correlation id
        String corr_id = (String) exchange.getProperty("org.apache.servicemix.correlationId");

        // node with date
        // Node parent = getExchangeBaseNode(session);
        Node parent = getMessageFlowBaseNode(session);

        // node with correlation id
        Node parent_corr = getCorrelationIdNode(parent, corr_id, id);

        try {
            return parent_corr.getNode(SlingUtils.getNodeCorrectName(id));

        } catch (PathNotFoundException e) {
            Node node = parent_corr.addNode(SlingUtils.getNodeCorrectName(id));
            node.setProperty(AuditContstants.RESOURCE_TYPE, AuditContstants.EXCHANGE_RESOURCE_TYPE);

            node.setProperty("exchangeId", id);
            node.setProperty("created", new DateValue(new GregorianCalendar()));

            node.addMixin("mix:versionable");
            LOG.debug("Created node " + node.getName());
            return node;
        }
    }

    /**
     * 
     * Get node for correlation id
     * 
     * @param parent
     * @param corr_id
     * @param id
     * @return Node
     * @throws RepositoryException
     * @throws ValueFormatException
     * @throws VersionException
     * @throws LockException
     * @throws ConstraintViolationException
     */
    private synchronized Node getCorrelationIdNode(Node parent, String corr_id, String id) throws RepositoryException,
            ValueFormatException, VersionException, LockException, ConstraintViolationException {

        // first exchange of the flow
        if (corr_id == null) {
            Node node = parent.addNode(SlingUtils.getNodeCorrectName(id));
            node.setProperty(AuditContstants.RESOURCE_TYPE, AuditContstants.MESSAGE_FLOW_RESOURCE_TYPE);

            node.setProperty("created", new DateValue(new GregorianCalendar()));
            node.setProperty("correlationId", id);
            return node;
        }

        try {
            return parent.getNode(SlingUtils.getNodeCorrectName(corr_id));

        } catch (PathNotFoundException e) {

            Node node = parent.addNode(SlingUtils.getNodeCorrectName(corr_id));
            node.setProperty(AuditContstants.RESOURCE_TYPE, AuditContstants.MESSAGE_FLOW_RESOURCE_TYPE);

            node.setProperty("created", new DateValue(new GregorianCalendar()));
            node.setProperty("correlationId", corr_id);

            return node;
        }
    }

    /**
     * 
     * Get base node for message flow
     * 
     * @param session
     * @return Node
     * @throws RepositoryException
     */
    private Node getMessageFlowBaseNode(Session session) throws RepositoryException {

        Node messageFlows = createOrGet(session.getRootNode(), AuditContstants.CONTENT_MESSAGE_FLOWS_TYPE,
                AuditContstants.MESSAGE_FLOWS_RESOURCE_TYPE);
        Node res = createOrGet(messageFlows, FORMAT.format(new Date()), AuditContstants.MESSAGE_FLOWS_RESOURCE_TYPE);
        session.save();     
        return res;
    }

    /**
     * 
     * Get base node for endpoint
     * 
     * @param session
     * @param endpointName
     * @return Node
     * @throws RepositoryException
     */
    private Node getEndpointNode(Session session, String endpointName) throws RepositoryException {

        Node endpoints = createOrGet(session.getRootNode(), AuditContstants.CONTENT_ENDPOINTS, AuditContstants.ENDPOINT_RESOURCE_TYPE);
        String id = SlingUtils.getNodeCorrectName(endpointName);
        Node res = createOrGet(endpoints, id, AuditContstants.ENDPOINT_RESOURCE_TYPE);
        session.save();
        return res;
    }

    /**
     * 
     * Create or get child node with path and resource type
     * 
     * @param parent
     * @param path
     * @param resourceType
     * @return Node
     * @throws ValueFormatException
     * @throws VersionException
     * @throws LockException
     * @throws ConstraintViolationException
     * @throws RepositoryException
     */
    private synchronized Node createOrGet(Node parent, String path, String resourceType) throws RepositoryException {
        Node node = RepositoryUtils.get(parent, path);
        if (node == null) {
            node = RepositoryUtils.getOrCreate(parent, path);
            node.setProperty(AuditContstants.RESOURCE_TYPE, resourceType);
            node.setProperty("created", new DateValue(new GregorianCalendar()));
            // node.setProperty("endpointId", (String) path);
            LOG.debug("Created node " + node.getName());
        }
        return node;
    }

    /**
     * 
     * Add in, out and fault messages, if exist
     * 
     * @param node
     * @param exchange
     * @throws ItemExistsException
     * @throws PathNotFoundException
     * @throws VersionException
     * @throws ConstraintViolationException
     * @throws LockException
     * @throws MessagingException
     * @throws RepositoryException
     * @throws TransformerException
     */
    private void addNormalizedMessages(Node node, MessageExchange exchange) throws ItemExistsException, PathNotFoundException,
            VersionException, ConstraintViolationException, LockException, MessagingException, RepositoryException, TransformerException {

        
        
        if (exchange.getMessage("in") != null) {
            addNormalizedMessages(node, "In", exchange.getMessage("in"));
        }
        if (exchange.getMessage("out") != null) {
            addNormalizedMessages(node, "Out", exchange.getMessage("out"));
        }
        if (exchange.getMessage("fault") != null) {
            addNormalizedMessages(node, "Fault", exchange.getMessage("fault"));
        }
    }

    /**
     * 
     * Add normalized message to node as property
     * 
     * @param parent
     * @param type
     * @param message
     * @throws ItemExistsException
     * @throws PathNotFoundException
     * @throws VersionException
     * @throws ConstraintViolationException
     * @throws LockException
     * @throws RepositoryException
     * @throws MessagingException
     * @throws TransformerException
     */
    private void addNormalizedMessages(Node parent, String type, NormalizedMessage message) throws ItemExistsException,
            PathNotFoundException, VersionException, ConstraintViolationException, LockException, RepositoryException, MessagingException,
            TransformerException {
        if (message != null) {
            Node node;
            try {
                node = parent.getNode(type);
            } catch (PathNotFoundException e) {
                node = parent.addNode(type);
            }
            node.setProperty("Content", getNormalizedMessageContent(message));
            for (Object key : message.getPropertyNames()) {
                String name = (String) key;
                node.setProperty(name, message.getProperty(name).toString());
            }
            node.setProperty(AuditContstants.RESOURCE_TYPE, AuditContstants.NORMALIZED_MESSAGE_RESOURCE_TYPE);
        }
    }

    /**
     * 
     * Get content from normalized message
     * 
     * @param message
     * @return String
     * @throws MessagingException
     * @throws TransformerException
     */
    private String getNormalizedMessageContent(NormalizedMessage message) throws MessagingException, TransformerException {
        MessageUtil.enableContentRereadability(message);
        return TRANSFORMER.toString(message.getContent());
    }
}
